package com.gmadmin.mq.listener;

import cn.hutool.json.JSONUtil;
import com.gmadmin.entity.dto.MessageNoticeDto;
import com.gmadmin.entity.vo.MessageInfoVo;
import com.gmadmin.mq.template.CustomMessageTemplate;
import com.gmadmin.utils.SecurityUtils;
import com.gmadmin.websocket.WsSessionManager;
import com.gmadmin.websocket.service.WsService;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 消息通知监听
 *
 * @author zhangliang
 * @version 1.0
 * @date 2022/8/4 12:53
 */
@RocketMQMessageListener(
        consumerGroup = "${rocketmq.namespace}%${rocketmq.consumer-notify.group}"  // 消费组，格式：namespace全称%group名称
        // 需要使用topic全称，所以进行topic名称的拼接，也可以自己设置  格式：namespace全称%topic名称
        , topic = "${rocketmq.namespace}%${rocketmq.consumer-notify.topic}"
        , selectorExpression = "${rocketmq.consumer-notify.subExpression}"
        , tlsEnable = "${rocketmq.consumer-notify.tlsEnable}" // 订阅表达式, 不配置表示订阅所有消息
)
@Service
public class MessageNoticeListener extends BaseMqMessageListener<MessageInfoVo>
        implements RocketMQListener<MessageInfoVo> {

    private final WsService wsService;


    protected MessageNoticeListener(CustomMessageTemplate customMessageTemplate, WsService wsService) {
        super(customMessageTemplate);
        this.wsService = wsService;
    }

    @Override
    protected String consumerName() {
        return "系统通知消费";
    }

    @Override
    protected void handleMessage(MessageInfoVo message) throws Exception {
// 全局通知
        if (message.getReceiveType() == 100) {
            wsService.broadcastMsg(JSONUtil.toJsonStr(message));
            return;
        }
        // 每个用户单独通知
        message.getMessageNotice().forEach(messageNoticeVo -> {
            try {
                wsService.sendMsg(WsSessionManager.get(String.valueOf(messageNoticeVo.getReceiveId())), JSONUtil.toJsonStr(message.getMessageNotice().stream().filter(noticeVo -> noticeVo.getReceiveId() == 1).findFirst().get()));
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        });
    }

    @Override
    protected void overMaxRetryTimesMessage(MessageInfoVo message) {
        // TODO document why this method is empty
    }

    @Override
    protected boolean isRetry() {
        return false;
    }

    @Override
    protected boolean isThrowException() {
        return false;
    }

    @Override
    public void onMessage(MessageInfoVo messageNoticeVo) {
        super.dispatchMessage(messageNoticeVo);
    }
}
